package com.heo.queue;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;

import com.heo.queue.task.CallBackTask;
import com.heo.queue.task.CallBackTaskThread;
import com.heo.queue.task.ReceiveTask;
import com.heo.queue.task.ReceiveTaskThread;
import com.heo.queue.task.SendTask;
import com.heo.queue.task.SendTaskThread;

public class TaskQueue
{
  private static LinkedBlockingQueue<SendTask> sendTaskQueue = new LinkedBlockingQueue<SendTask>();// 给设备发送命令的队列
  private static LinkedBlockingQueue<ReceiveTask> receiveTaskQueue = new LinkedBlockingQueue<ReceiveTask>();// 收到设备消息的队列
  private static LinkedBlockingQueue<CallBackTask> callbackTaskQueue = new LinkedBlockingQueue<CallBackTask>();// 回调消息队列
  private static final int MAX_ACTIVE_THREAD_COUNT = 200;// 默认最大的线程数量
  // 线程池
  private static ExecutorService sendTaskThreadPool = Executors.newFixedThreadPool(MAX_ACTIVE_THREAD_COUNT);// 发送任务线程池
  private static ExecutorService receiveTaskThreadPool = Executors.newFixedThreadPool(MAX_ACTIVE_THREAD_COUNT);// 接收任务线程池
  private static ExecutorService callBackTaskThreadPool = Executors.newFixedThreadPool(MAX_ACTIVE_THREAD_COUNT);// 回调任务线程池
  

  /**
   * 启动发送数据包任务
   */
  private static void startSendTask()
  {
    sendTaskThreadPool.execute(new SendTaskThread());
  }

  /**
   * 启动发送回调消息任务
   */
  private static void startCallBackTask()
  {
    callBackTaskThreadPool.execute(new CallBackTaskThread());
  }

  /**
   * 启动发送数据包任务
   */
  private static void startReceiveTask()
  {
    receiveTaskThreadPool.execute(new ReceiveTaskThread());
  }

  /**
   * 从发送队列中，取出一个需要发送消息的数据 不会阻塞线程，也不会抛出异常
   * 
   * @return
   */
  public static SendTask pollSendTask()
  {
    return sendTaskQueue.poll();
  }

  /**
   * 取出一个回调消息
   * 
   * @return
   */
  public static CallBackTask pollCallBackTask()
  {
    return callbackTaskQueue.poll();
  }

  /**
   * 从发送队列中，取出一个需要发送消息的数据，如果当前队列为null就阻塞 会阻塞线程
   * 
   * @return
   * @throws InterruptedException
   */
  public static SendTask takeSendTask() throws InterruptedException
  {
    return sendTaskQueue.take();
  }

  /**
   * 将一个将收到的指令放入队列 不会阻塞线程，也不会抛出异常 当队列被元素填满后，再调用offer(e)，则不会插入元素，函数返回false。
   * 
   * @param sendTask
   * @param isStartTask
   * @return
   */
  public static boolean offerSendTask(SendTask sendTask, boolean isStartTask)
  {
    boolean result = sendTaskQueue.offer(sendTask);
    if (isStartTask)
    {
      int threadCount = ((ThreadPoolExecutor) sendTaskThreadPool).getActiveCount();
      if (threadCount < MAX_ACTIVE_THREAD_COUNT)
      {
        startSendTask();
      }
    }
    return result;
  }

  /**
   * 把回调消息任务放入队列
   * 
   * @param callBackTask
   * @return
   */
  public static boolean offerCallBackTask(CallBackTask callBackTask, boolean isStartTask)
  {
    boolean result = callbackTaskQueue.offer(callBackTask);
    
    if (isStartTask)
    {
      int threadCount = ((ThreadPoolExecutor) callBackTaskThreadPool).getActiveCount();
      if (threadCount < MAX_ACTIVE_THREAD_COUNT)
      {
        startCallBackTask();
      }
    }
    return result;
  }

  /**
   * 将一个将收到的指令放入队列 会阻塞线程
   * 
   * @return
   * @throws InterruptedException
   */
  public static void putSendTask(SendTask sendTask) throws InterruptedException
  {
    sendTaskQueue.put(sendTask);
    startSendTask();
  }

  public static int getSendTaskSize()
  {
    return sendTaskQueue.size();
  }

  /**
   * 从接收队列中，取出一个需要发送消息的数据 不会阻塞线程，也不会抛出异常
   * 
   * @return
   */
  public static ReceiveTask pollReceiveTask()
  {
    return receiveTaskQueue.poll();
  }

  /**
   * 从接收队列中，取出一个需要发送消息的数据 会阻塞线程
   * 
   * @return
   * @throws InterruptedException
   */
  public static ReceiveTask takeReceiveTask() throws InterruptedException
  {
    return receiveTaskQueue.take();
  }

  /**
   * 将一个将收到的指令放入队列 不会阻塞线程，也不会抛出异常 当队列被元素填满后，再调用offer(e)，则不会插入元素，函数返回false。
   * 
   * @return
   */
  public static boolean offerReceiveTask(ReceiveTask receiveTask, boolean isStartTask)
  {
    boolean result = receiveTaskQueue.offer(receiveTask);
    
    if (isStartTask)
    {
      int threadCount = ((ThreadPoolExecutor) receiveTaskThreadPool).getActiveCount();
      if (threadCount < MAX_ACTIVE_THREAD_COUNT)
      {
        startReceiveTask();
      }
    }

    return result;
  }

  /**
   * 将一个将收到的指令放入队列 会阻塞线程
   * 
   * @return
   * @throws InterruptedException
   */
  public static void putReceiveTask(ReceiveTask receiveTask) throws InterruptedException
  {
    receiveTaskQueue.put(receiveTask);
  }

  public static int getReceiveTaskSize()
  {
    return receiveTaskQueue.size();
  }

  public static int getCallBackTaskSize()
  {
    return callbackTaskQueue.size();
  }
}
